package com.lagou.mycep;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

public class TempDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStreamSource<PayBean> data = env.fromElements(
                new PayBean(1L, "create", 1614742775826L), //2021-03-03 11:39:35
                new PayBean(1L, "pay", 1614742794836L), //2021-03-03 11:39:54
                new PayBean(2L, "create", 1614742794836L)
        );

        SingleOutputStreamOperator<PayBean> watermarks = data.assignTimestampsAndWatermarks(new WatermarkStrategy<PayBean>() {
            @Override
            public WatermarkGenerator<PayBean> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<PayBean>() {
                    Long maxtimestamp = 0L;
                    Long maxOutOfOrderness = 500L;

                    @Override
                    public void onEvent(PayBean event, long eventTimestamp, WatermarkOutput output) {
                        maxtimestamp = Math.max(maxtimestamp, event.getTs());
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(maxtimestamp - maxOutOfOrderness));
                    }
                };
            }
        }.withTimestampAssigner(new SerializableTimestampAssigner<PayBean>() {
            @Override
            public long extractTimestamp(PayBean element, long recordTimestamp) {
                return element.getId();
            }
        }));

        KeyedStream<PayBean, Long> keyed = watermarks.keyBy(x -> x.getId());

        Pattern<PayBean, PayBean> myPattern = Pattern.<PayBean>begin("start").where(new IterativeCondition<PayBean>() {
            @Override
            public boolean filter(PayBean value, Context<PayBean> ctx) throws Exception {
                return value.getState().equals("create");
            }
        })
                .followedBy("next").where(new IterativeCondition<PayBean>() {
                    @Override
                    public boolean filter(PayBean value, Context<PayBean> ctx) throws Exception {
                        return value.getState().equals("pay");
                    }
                })
                .within(Time.seconds(10));

        PatternStream<PayBean> patternStream = CEP.pattern(keyed, myPattern);

        OutputTag<PayBean> myOutputTag = new OutputTag<PayBean>("myOutputTag") {
        };

        SingleOutputStreamOperator<PayBean> selected = patternStream.select(
                myOutputTag,
                new PatternTimeoutFunction<PayBean, PayBean>() {
                    @Override
                    public PayBean timeout(Map<String, List<PayBean>> pattern, long timeoutTimestamp) throws Exception {
                        System.out.println("timeout: " + pattern);
                        return pattern.get("start").get(0);
                    }
                },
                new PatternSelectFunction<PayBean, PayBean>() {
                    @Override
                    public PayBean select(Map<String, List<PayBean>> pattern) throws Exception {
                        System.out.println("select: " + pattern);
                        return pattern.get("start").get(0);
                    }
                });
        DataStream<PayBean> sideOutput = selected.getSideOutput(myOutputTag);
        sideOutput.print();

        env.execute();

    }
}
